package discard.kv;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import hyl.base.cache.redis.CRedis;
import hyl.base.cache.redis.RedisPool;
import hyl.core.MyFun;
import hyl.core.data.ICache;
import hyl.core.data.ICacheConfig;
import hyl.core.run.MyTimer;
import redis.clients.jedis.Pipeline;

public class ConfigCache implements ICacheConfig {
	/**
	 * 缓存启动动第 n 毫秒开始自动运行保存程序 默认延时20ms后启动
	 */
	public int DELAY_MS = 20;
	/**
	 * 内存数据每隔多久自动保存到redis一次,间隔时间 :n 毫秒
	 */
	public int PERIOD_MS = 500;
	/**
	 * redis数据库索引位置
	 */
	public int Redis_DB = CRedis.IDX_SESSION;
	/**
	 * redis超时,超时后自动删除 大于0时有效,<=0时不删除
	 */
	public Long EXPIRE_MS = -1L;
	/**
	 * 用于 map 类型的 cache 时, 限定 map的key 的多少 用于 list queue set 集合时 ,限定每一组 key的 field 数量
	 */
	public int MAX_KEYS = 60000;
	/**
	 * 默认满100条立即插入
	 */
	public int I满N条后插入 = 0;
	/**
	 * redis数据库 默认别名
	 */
	public String Redis_alias = "redis1";
	public AtomicInteger _commitcount = new AtomicInteger(0);
	private List<ICache> caches = new ArrayList<>();
	// 操作指令
	public static final char CTypeString = 'S';
	public static final char CTypeBytes = 'B';
	public static final char Chset = 'A';
	public static final char Chdel = 'B';
	public static final char Cdel = 'C';
	public static final char Clpush = 'D';
	public static final char Crpush = 'E';
	public static final char Clpop = 'F';
	public static final char Crpop = 'G';

	/**
	 * 添加缓存
	 */
	public void AddCache(ICache a) {
		caches.add(a);
	}

	/**
	 * 移除缓存
	 */
	public void removeCache(ICache a) {
		caches.remove(a);
	}

	/**
	 * 延时启动常数，留出准备时间
	 */
	public void setCache初始化时长(String 毫秒) {
		if (MyFun.isEmpty(毫秒))
			return;
		DELAY_MS = MyFun.str2int(毫秒);
	}

	/**
	 * 缓存过期时间常数 <0 时不删除
	 */
	public void set数据过期时长(String 秒) {
		if (MyFun.isEmpty(秒))
			return;
		EXPIRE_MS = MyFun.str2long(秒) * 1000;
	}

	/**
	 * 提交间隔时间常数
	 */
	public void set提交时间间隔(String 毫秒) {
		if (MyFun.isEmpty(毫秒))
			return;
		PERIOD_MS = MyFun.str2int(毫秒);
	}

	public int getMAX_KEYS() {
		return MAX_KEYS;
	}

	public void set每组缓存最大键数(String b) {
		if (MyFun.isEmpty(b))
			return;
		MAX_KEYS = MyFun.str2int(b);
	}

	public void set满N条提交(String b) {
		if (MyFun.isEmpty(b))
			return;
		I满N条后插入 = MyFun.str2int(b);
	}

	/**
	 * 加载配置
	 */
	protected ConfigCache(Map<String, String> conf) {

		setCache初始化时长(conf.get("rd_delays"));
		set提交时间间隔(conf.get("rd_periods"));
		set数据过期时长(conf.get("ttl"));
		set每组缓存最大键数(conf.get("maxkeys"));// 可保存最大数量 只适用于Iqueue
		set满N条提交(conf.get("minnum"));// 可保存最大数量 只适用于Iqueue
		setDb(conf.get("rd_alias"), conf.get("rd_db"));
		getPipeline();
		timeoutCommit();
	}

	private void setDb(String rd, String db) {
		if (!MyFun.isEmpty(db))
			Redis_DB = MyFun.str2int(db);
		if (!MyFun.isEmpty(rd)) {
			Redis_alias = rd;
		}
	}

	protected Pipeline _pl = null;
	protected CRedis _redis;

	public synchronized CRedis getRedis() {
		if (_redis == null) {
			_redis = (CRedis) RedisPool.getInstance().getRedis(Redis_alias, Redis_DB);
		}
//		if(!_redis.getJedis().isConnected())
//			_redis.getJedis().connect();
		return _redis;
	}

	private synchronized Pipeline getPipeline() {
		if (_pl == null) {
			getRedis();
			_pl = _redis.createPipeline();
		}
		return _pl;
	}

	/* 把提交命令放入 队列 */
	protected ConcurrentLinkedQueue<Object[]> opCommands = new ConcurrentLinkedQueue<>();

	/**
	 * 操作redis 指令 插入opCommands 队列
	 * 
	 * @param record
	 */
	public synchronized void addRedisCommand(Object[] record) {
		opCommands.offer(record);
		_commitcount.incrementAndGet();
	}

	public synchronized void commit批量(int max) {
		if (_commitcount.get() > max) {
			if (_pl != null) {
				while (!opCommands.isEmpty()) {
					Object[] cmd = opCommands.poll();
					execPipelineCommand(cmd);
				}
				_pl.sync();
			}
			_commitcount.set(0);
		}
	}

	// 提交方法
	/**
	 * 手动提交
	 */
	public synchronized void commit() {
		commit批量(0);
	}

	/**
	 * 批量提交
	 */
	public synchronized void commit批量() {
		commit批量(I满N条后插入);
	}

	/**
	 * 执行 redis Pipeline 指令
	 * 
	 * @param cmd
	 */
	public void execPipelineCommand(Object[] cmd) {
		char c = (char) cmd[0];
		char t = (char) cmd[1];
		if (t == CTypeString) {
			String group = (String) cmd[2];
			switch (c) {
			case Chset:
				_pl.hset(group, (String) cmd[3], (String) cmd[4]);
				break;
			case Chdel:
				_pl.hdel(group, (String) cmd[3]);
				break;
			case Cdel:
				_pl.del(group);
				break;
			case Clpush:
				_pl.lpush(group, (String) cmd[3]);
				break;
			case Crpush:
				_pl.rpush(group, (String) cmd[3]);
				break;
			case Clpop:
				_pl.lpop(group);
				break;
			case Crpop:
				_pl.rpop(group);
				break;
			}
		} else {
			byte[] group = (byte[]) cmd[2];
			switch (c) {
			case Chset:
				_pl.hset(group, (byte[]) cmd[3], (byte[]) cmd[4]);
				break;
			case Chdel:
				_pl.hdel(group, (byte[]) cmd[3]);
				break;
			case Cdel:
				_pl.del(group);
				break;
			case Clpush:
				_pl.lpush(group, (byte[]) cmd[3]);
				break;
			case Crpush:
				_pl.rpush(group, (byte[]) cmd[3]);
				break;
			case Clpop:
				_pl.lpop(group);
				break;
			case Crpop:
				_pl.rpop(group);
				break;
			}
		}
	}

	private MyTimer _mtimer = null;
    public void close() {
    	_mtimer.close();
    	_redis.close();
    }
	// 每隔一段时间自动提交
	public synchronized void timeoutCommit() {
		// Schedule.close("定时线程：缓冲池");

		if (_mtimer == null) {
			_mtimer = MyTimer.getInstance();
			// System.out.println("cache.ConfigCache.timeoutCommit()=" + _mtimer.getName());
			_mtimer.start(new Runnable() {
				@Override
				public void run() {
					// 提交
					commit();
					/*
					 * 如果设定了缓存过期,那么超时后自动删除,否则不删除 判断过期后自动清除,清除时还会调用commit函数;
					 */
					if (EXPIRE_MS > 0) {
						for (ICache a : caches) {
							a.expire();
						}
					}
				}
			}, (long) DELAY_MS, (long) PERIOD_MS);
		}
	}

}
